package com.wulaobo.test;

import com.wulaobo.pojo.Event;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class SourceCustom {

    public static void main(String[] args) throws Exception {
        //1.根据当前运行的上下文,获取正确的执行环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //2.获取数据源，读取数据
        DataStreamSource<Event> eventDataStreamSource = env.addSource(new ClickSource());

        //3.进行转换计算
        //一个 Flink 程序的核心，其实就是所有的转换操作，它们决定了处理的业务逻辑。
        SingleOutputStreamOperator<String> operator = eventDataStreamSource
                .map(event -> event.getUser()+"=="+event.getUrl());

        //4.打印输出
        operator.print();

        //触发程序执行
        //Flink 是由事件驱动的，只有等到数据到来，才会触发真正的计算，
        //这也被称为“延迟执行”或“懒执行”（lazy execution）。
        //所以我们需要显式地调用执行环境的 execute()方法，来触发程序执行。
        env.execute();

    }

}
